package com.xiaochao.bigdata.spark.session
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import com.xiaochao.bigdata.conf.ConfigurationManager
import com.xiaochao.bigdata.constant.Constants
import com.xiaochao.bigdata.dao.factory.DAOFactory
import com.xiaochao.bigdata.domain.SessionAggrStat
import com.xiaochao.bigdata.spark.accumulator.SessionAggrStatAccumulator
import com.xiaochao.bigdata.test.MockData
import com.xiaochao.bigdata.util._
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SQLContext, SparkSession}

/**
  * 用户访问session分析Spark作业
  *
  * 接收用户创建的分析任务，用户可能指定的条件如下：
  *
  * 1、时间范围：起始日期~结束日期
  * 2、性别：男或女
  * 3、年龄范围
  * 4、职业：多选
  * 5、城市：多选
  * 6、搜索词：多个搜索词，只要某个session中的任何一个action搜索过指定的关键词，那么session就符合条件
  * 7、点击品类：多个品类，只要某个session中的任何一个action点击过某个品类，那么session就符合条件
  *
  * 我们的spark作业如何接受用户创建的任务？
  *
  * J2EE平台在接收用户创建任务的请求之后，会将任务信息插入MySQL的task表中，任务参数以JSON格式封装在task_param
  * 字段中
  *
  * 接着J2EE平台会执行我们的spark-submit shell脚本，并将taskid作为参数传递给spark-submit shell脚本
  * spark-submit shell脚本，在执行时，是可以接收参数的，并且会将接收的参数，传递给Spark作业的main函数
  * 参数就封装在main函数的args数组中
  *
  * 这是spark本身提供的特性
  *
  * @author Administrator
  *
  */
object UserVisitSessionAnalyzeSpark {


  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .appName(Constants.SPARK_APP_NAME_SESSION)
      .master("local")
      .enableHiveSupport()
      .getOrCreate()

    //模拟数据

    mockData(spark.sparkContext,spark.sqlContext)

//    val taskId = ParamUtils.getTaskIdFromArgs(args)
    // 首先得查询出来指定的任务，并获取任务的查询参数

    val taskDAO = DAOFactory.getTaskDAO()

    val task = taskDAO.findById(1)

    val taskParam:JSONObject = JSON.parseObject(task.getTaskParam)

    // 如果要进行session粒度的数据聚合
    // 首先要从user_visit_action表中，查询出来指定日期范围内的行为数据
    val actionRDD = getActionRDDByDateRange(spark.sqlContext, taskParam)


    val sessionid2FullAggrInfoRDD = aggregateBySession(spark.sqlContext,actionRDD)

    // 使用accumulator()()方法（curry），创建自定义的Accumulator
    val sessionAggrStatAccumulator = new SessionAggrStatAccumulator()
    spark.sparkContext.register(sessionAggrStatAccumulator,"sessionAggrStatAccumulator")

    // 按参数过滤

    val filteredSessionid2AggrInfoRDD = filterSession(spark.sqlContext,sessionid2FullAggrInfoRDD,taskParam,sessionAggrStatAccumulator)



    /**
      * 对于Accumulator这种分布式累加计算的变量的使用，有一个重要说明
      *
      * 从Accumulator中，获取数据，插入数据库的时候，一定要，一定要，是在有某一个action操作以后
      * 再进行。。。
      *
      * 如果没有action的话，那么整个程序根本不会运行。。。
      *
      * 是不是在calculateAndPersisitAggrStat方法之后，运行一个action操作，比如count、take
      * 不对！！！
      *
      * 必须把能够触发job执行的操作，放在最终写入MySQL方法之前
      *
      * 计算出来的结果，在J2EE中，是怎么显示的，是用两张柱状图显示
      */

    println(filteredSessionid2AggrInfoRDD.count())

    // 计算出各个范围的session占比，并写入MySQL
    calculateAndPersistAggrStat(sessionAggrStatAccumulator.value,task.getTaskid)


    /**
      * session聚合统计（统计出访问时长和访问步长，各个区间的session数量占总session数量的比例）
      *
      * 如果不进行重构，直接来实现，思路：
      * 1、actionRDD，映射成<sessionid,Row>的格式
      * 2、按sessionid聚合，计算出每个session的访问时长和访问步长，生成一个新的RDD
      * 3、遍历新生成的RDD，将每个session的访问时长和访问步长，去更新自定义Accumulator中的对应的值
      * 4、使用自定义Accumulator中的统计值，去计算各个区间的比例
      * 5、将最后计算出来的结果，写入MySQL对应的表中
      *
      * 普通实现思路的问题：
      * 1、为什么还要用actionRDD，去映射？其实我们之前在session聚合的时候，映射已经做过了。多此一举
      * 2、是不是一定要，为了session的聚合这个功能，单独去遍历一遍session？其实没有必要，已经有session数据
      * 		之前过滤session的时候，其实，就相当于，是在遍历session，那么这里就没有必要再过滤一遍了
      *
      * 重构实现思路：
      * 1、不要去生成任何新的RDD（处理上亿的数据）
      * 2、不要去单独遍历一遍session的数据（处理上千万的数据）
      * 3、可以在进行session聚合的时候，就直接计算出来每个session的访问时长和访问步长
      * 4、在进行过滤的时候，本来就要遍历所有的聚合session信息，此时，就可以在某个session通过筛选条件后
      * 		将其访问时长和访问步长，累加到自定义的Accumulator上面去
      * 5、就是两种截然不同的思考方式，和实现方式，在面对上亿，上千万数据的时候，甚至可以节省时间长达
      * 		半个小时，或者数个小时
      *
      * 开发Spark大型复杂项目的一些经验准则：
      * 1、尽量少生成RDD
      * 2、尽量少对RDD进行算子操作，如果有可能，尽量在一个算子里面，实现多个需要做的功能
      * 3、尽量少对RDD进行shuffle算子操作，比如groupByKey、reduceByKey、sortByKey（map、mapToPair）
      * 		shuffle操作，会导致大量的磁盘读写，严重降低性能
      * 		有shuffle的算子，和没有shuffle的算子，甚至性能，会达到几十分钟，甚至数个小时的差别
      * 		有shfufle的算子，很容易导致数据倾斜，一旦数据倾斜，简直就是性能杀手（完整的解决方案）
      * 4、无论做什么功能，性能第一
      * 		在传统的J2EE或者.NET后者PHP，软件/系统/网站开发中，我认为是架构和可维护性，可扩展性的重要
      * 		程度，远远高于了性能，大量的分布式的架构，设计模式，代码的划分，类的划分（高并发网站除外）
      *
      * 		在大数据项目中，比如MapReduce、Hive、Spark、Storm，我认为性能的重要程度，远远大于一些代码
      * 		的规范，和设计模式，代码的划分，类的划分；大数据，大数据，最重要的，就是性能
      * 		主要就是因为大数据以及大数据项目的特点，决定了，大数据的程序和项目的速度，都比较慢
      * 		如果不优先考虑性能的话，会导致一个大数据处理程序运行时间长度数个小时，甚至数十个小时
      * 		此时，对于用户体验，简直就是一场灾难
      *
      * 		所以，推荐大数据项目，在开发和代码的架构中，优先考虑性能；其次考虑功能代码的划分、解耦合
      *
      * 		我们如果采用第一种实现方案，那么其实就是代码划分（解耦合、可维护）优先，设计优先
      * 		如果采用第二种方案，那么其实就是性能优先
      *
      * 		讲了这么多，其实大家不要以为我是在岔开话题，大家不要觉得项目的课程，就是单纯的项目本身以及
      * 		代码coding最重要，其实项目，我觉得，最重要的，除了技术本身和项目经验以外；非常重要的一点，就是
      * 		积累了，处理各种问题的经验
      *
      */




  }


  /**
    * 获取指定日期范围内的用户访问行为数据
    * @param sqlContext SQLContext
    * @param taskParam 任务参数
    * @return 行为数据RDD
    */

  private def getActionRDDByDateRange(sqlContext:SQLContext,taskParam:JSONObject):RDD[Row]= {

    val startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE);
    val endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE);

    val sql = "select * " + "from user_visit_action " + "where date>='" + startDate + "' " + "and date<='" + endDate + "'"

    val actionDF = sqlContext.sql(sql);

    actionDF.rdd


  }
  /**
    * 对行为数据按session粒度进行聚合
    * @param actionRDD 行为数据RDD
    * @return session粒度聚合数据
    */
  private def aggregateBySession(sqlContext: SQLContext,actionRDD: RDD[Row]):RDD[(String, String)]={

    // 现在actionRDD中的元素是Row，一个Row就是一行用户访问行为记录，比如一次点击或者搜索
    // 我们现在需要将这个Row映射成<sessionid,Row>的格式
    val sessionid2ActionRDD = actionRDD.map((row:Row)=>{

      (row.getString(2),row)

    })
    // 对行为数据按session粒度进行分组
    val sessionid2ActionsRDD =
      sessionid2ActionRDD.groupByKey()
    println("sessionid2ActionsRDD:"+sessionid2ActionsRDD.count())

    // 对每一个session分组进行聚合，将session中所有的搜索词和点击品类都聚合起来
    // 到此为止，获取的数据格式，如下：如下：<userid,partAggrInfo(sessionid,searchKeywords,clickCategoryIds)>

    val userid2PartAggrInfoRDD = sessionid2ActionsRDD.map((sessionTuple:(String,Iterable[Row]))=>{

      val sessionid = sessionTuple._1;
      val iterator = sessionTuple._2.iterator;

      val searchKeywordsBuffer = new StringBuffer("")
      val clickCategoryIdsBuffer = new StringBuffer("")

      var userid = ""

      // session的起始和结束时间
      var startTime:Date = null
      var endTime:Date = null
      // session的访问步长
      var stepLength = 0




      // 遍历session所有的访问行为
      while(iterator.hasNext) {
        // 提取每个访问行为的搜索词字段和点击品类字段
        val row = iterator.next()
        if(userid.length == 0) {
          userid = row.getString(1)
        }

        val searchKeyword = row.getString(5)
        val clickCategoryId = row.getString(6)

        // 实际上这里要对数据说明一下
        // 并不是每一行访问行为都有searchKeyword何clickCategoryId两个字段的
        // 其实，只有搜索行为，是有searchKeyword字段的
        // 只有点击品类的行为，是有clickCategoryId字段的
        // 所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的

        // 我们决定是否将搜索词或点击品类id拼接到字符串中去
        // 首先要满足：不能是null值
        // 其次，之前的字符串中还没有搜索词或者点击品类id

        if(StringUtils.isNotEmpty(searchKeyword)) {
          if(!searchKeywordsBuffer.toString().contains(searchKeyword)) {
            searchKeywordsBuffer.append(searchKeyword + ",")
          }
        }
        if(clickCategoryId != null) {
          if(!clickCategoryIdsBuffer.toString().contains(
            String.valueOf(clickCategoryId))) {
            clickCategoryIdsBuffer.append(clickCategoryId + ",")
          }
        }

        // 计算session开始和结束时间
        val actionTime = DateUtils.parseTime(row.getString(4))

        if(startTime == null) {
          startTime = actionTime
        }
        if(endTime == null) {
          endTime = actionTime
        }

        if(actionTime.before(startTime)) {
          startTime = actionTime
        }
        if(actionTime.after(endTime)) {
          endTime = actionTime
        }

        // 计算session访问步长
        stepLength +=1


      }

      val searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString())
      val clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString())
      // 计算session访问时长（秒）
      val visitLength = (endTime.getTime() - startTime.getTime()) / 1000

      // 大家思考一下
      // 我们返回的数据格式，即使<sessionid,partAggrInfo>
      // 但是，这一步聚合完了以后，其实，我们是还需要将每一行数据，跟对应的用户信息进行聚合
      // 问题就来了，如果是跟用户信息进行聚合的话，那么key，就不应该是sessionid
      // 就应该是userid，才能够跟<userid,Row>格式的用户信息进行聚合
      // 如果我们这里直接返回<sessionid,partAggrInfo>，还得再做一次mapToPair算子
      // 将RDD映射成<userid,partAggrInfo>的格式，那么就多此一举

      // 所以，我们这里其实可以直接，返回的数据格式，就是<userid,partAggrInfo>
      // 然后跟用户信息join的时候，将partAggrInfo关联上userInfo
      // 然后再直接将返回的Tuple的key设置成sessionid
      // 最后的数据格式，还是<sessionid,fullAggrInfo>

      // 聚合数据，用什么样的格式进行拼接？
      // 我们这里统一定义，使用key=value|key=value

      val partAggrInfo = Constants.FIELD_SESSION_ID+ "=" + sessionid + "|" + Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKeywords + "|" + Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCategoryIds +"|" + Constants.FIELD_VISIT_LENGTH + "=" + visitLength +"|" + Constants.FIELD_STEP_LENGTH + "=" + stepLength

      (userid,partAggrInfo)


    });

    // 查询所有用户数据，并映射成<userid,Row>的格式
    val sql = "select * from user_info"

    val userInfoRDD = sqlContext.sql(sql).rdd

    val userid2InfoRDD:RDD[(String,Row)] = userInfoRDD.map((row: Row)=>{


      (row.getString(0),row)

    })

    // 将session粒度聚合数据，与用户信息进行join

    val userid2FullInfoRDD =
      userid2PartAggrInfoRDD.join(userid2InfoRDD)

    // 对join起来的数据进行拼接，并且返回<sessionid,fullAggrInfo>格式的数据

    val sessionid2FullAggrInfoRDD = userid2FullInfoRDD.map((tuple:(String,(String,Row)))=>{

      val partAggrInfo = tuple._2._1
      val userInfoRow = tuple._2._2

      val sessionid = StringUtils.getFieldFromConcatString(
        partAggrInfo, "\\|", Constants.FIELD_SESSION_ID)

      val age = userInfoRow.getInt(3)
      val professional = userInfoRow.getString(4)
      val city = userInfoRow.getString(5)
      val sex = userInfoRow.getString(6)

      val fullAggrInfo:String = partAggrInfo + "|" + Constants.FIELD_AGE + "=" + age + "|" + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"+ Constants.FIELD_CITY + "=" + city + "|"+ Constants.FIELD_SEX + "=" + sex


      (sessionid,fullAggrInfo)

    })

    sessionid2FullAggrInfoRDD
  }

  /**
    * 过滤session数据
    * @param sessionid2AggrInfoRDD
    * @return
    */
  private def filterSession(sqlContext: SQLContext,sessionid2AggrInfoRDD:RDD[(String,String)],taskParam:JSONObject,sessionAggrStatAccumulator:SessionAggrStatAccumulator):RDD[(String,String)]= {


    // 为了使用我们后面的ValieUtils，所以，首先将所有的筛选参数拼接成一个连接串
    // 此外，这里其实大家不要觉得是多此一举
    // 其实我们是给后面的性能优化埋下了一个伏笔
    val startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE)
    val endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE)
    val professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS)
    val cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES)
    val sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX)
    val keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS)
    val categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS)

    var _parameter = (if (startAge != null) Constants.PARAM_START_AGE + "=" + startAge + "|"
    else "") + (if (endAge != null) Constants.PARAM_END_AGE + "=" + endAge + "|"
    else "") + (if (professionals != null) Constants.PARAM_PROFESSIONALS + "=" + professionals + "|"
    else "") + (if (cities != null) Constants.PARAM_CITIES + "=" + cities + "|"
    else "") + (if (sex != null) Constants.PARAM_SEX + "=" + sex + "|"
    else "") + (if (keywords != null) Constants.PARAM_KEYWORDS + "=" + keywords + "|"
    else "") + (if (categoryIds != null) Constants.PARAM_CATEGORY_IDS + "=" + categoryIds
    else "")

    if (_parameter.endsWith("|")) _parameter = _parameter.substring(0, _parameter.length - 1)
    val parameter = _parameter

    // 根据筛选参数进行过滤
    val filteredSessionid2AggrInfoRDD = sessionid2AggrInfoRDD.filter((tuple:(String,String))=>{

      val aggrInfo = tuple._2
      // 接着，依次按照筛选条件进行过滤
      // 按照年龄范围进行过滤（startAge、endAge）
      val flag = (!ValidUtils.between(aggrInfo, Constants.FIELD_AGE, parameter, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) || (!ValidUtils.in(aggrInfo, Constants.FIELD_PROFESSIONAL, parameter, Constants.PARAM_PROFESSIONALS))||(!ValidUtils.in(aggrInfo, Constants.FIELD_CITY, parameter, Constants.PARAM_CITIES)) || (!ValidUtils.equal(aggrInfo, Constants.FIELD_SEX, parameter, Constants.PARAM_SEX)) || (!ValidUtils.in(aggrInfo, Constants.FIELD_SEARCH_KEYWORDS, parameter, Constants.PARAM_KEYWORDS)) || (!ValidUtils.in(aggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS, parameter, Constants.PARAM_CATEGORY_IDS))

      // 如果经过了之前的多个过滤条件之后，程序能够走到这里
      // 那么就说明，该session是通过了用户指定的筛选条件的，也就是需要保留的session
      // 那么就要对session的访问时长和访问步长，进行统计，根据session对应的范围
      // 进行相应的累加计数

      // 主要走到这一步，那么就是需要计数的session
      sessionAggrStatAccumulator.add(Constants.SESSION_COUNT)

      // 计算出session的访问时长和访问步长的范围，并进行相应的累加
      val visitLength:Long = (StringUtils.getFieldFromConcatString(
        aggrInfo, "\\|", Constants.FIELD_VISIT_LENGTH)).toLong
      val stepLength:Long = (StringUtils.getFieldFromConcatString(
        aggrInfo, "\\|", Constants.FIELD_STEP_LENGTH)).toLong
      calculateVisitLength(visitLength,sessionAggrStatAccumulator)
      calculateStepLength(stepLength,sessionAggrStatAccumulator)

      !flag
    })

    filteredSessionid2AggrInfoRDD

  }

  /**
    * 计算访问时长范围
    *
    * @param visitLength
    */
  private def calculateVisitLength(visitLength: Long,sessionAggrStatAccumulator:SessionAggrStatAccumulator): Unit = {
    if (visitLength >= 1 && visitLength <= 3) sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1s_3s)
    else if (visitLength >= 4 && visitLength <= 6) sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_4s_6s)
    else if (visitLength >= 7 && visitLength <= 9) sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_7s_9s)
    else if (visitLength >= 10 && visitLength <= 30) sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10s_30s)
    else if (visitLength > 30 && visitLength <= 60) sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30s_60s)
    else if (visitLength > 60 && visitLength <= 180) sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1m_3m)
    else if (visitLength > 180 && visitLength <= 600) sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_3m_10m)
    else if (visitLength > 600 && visitLength <= 1800) sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10m_30m)
    else if (visitLength > 1800) sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30m)
  }

  /**
    * 计算访问步长范围
    *
    * @param stepLength
    */
  private def calculateStepLength(stepLength: Long,sessionAggrStatAccumulator:SessionAggrStatAccumulator): Unit = {
    if (stepLength >= 1 && stepLength <= 3) sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_1_3)
    else if (stepLength >= 4 && stepLength <= 6) sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_4_6)
    else if (stepLength >= 7 && stepLength <= 9) sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_7_9)
    else if (stepLength >= 10 && stepLength <= 30) sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_10_30)
    else if (stepLength > 30 && stepLength <= 60) sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_30_60)
    else if (stepLength > 60) sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_60)
  }



  /**
    * 计算各session范围占比，并写入MySQL
    *
    * @param value
    */
  private def calculateAndPersistAggrStat(value: String, taskid: Long): Unit = { // 从Accumulator统计串中获取值

    val session_count = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.SESSION_COUNT)).toLong
    val visit_length_1s_3s = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_1s_3s)).toDouble
    val visit_length_4s_6s = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_4s_6s)).toDouble
    val visit_length_7s_9s = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_7s_9s)).toDouble
    val visit_length_10s_30s = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_10s_30s)).toDouble
    val visit_length_30s_60s = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_30s_60s)).toDouble
    val visit_length_1m_3m = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_1m_3m)).toDouble
    val visit_length_3m_10m = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_3m_10m)).toDouble
    val visit_length_10m_30m = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_10m_30m)).toDouble
    val visit_length_30m = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.TIME_PERIOD_30m)).toDouble
    val step_length_1_3 = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_1_3)).toDouble
    val step_length_4_6 = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_4_6)).toDouble
    val step_length_7_9 = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_7_9)).toDouble
    val step_length_10_30 = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_10_30)).toDouble
    val step_length_30_60 = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_30_60)).toDouble
    val step_length_60 = (StringUtils.getFieldFromConcatString(value, "\\|", Constants.STEP_PERIOD_60)).toDouble
    // 计算各个访问时长和访问步长的范围
    val visit_length_1s_3s_ratio = NumberUtils.formatDouble(visit_length_1s_3s / session_count, 2)
    val visit_length_4s_6s_ratio = NumberUtils.formatDouble(visit_length_4s_6s / session_count, 2)
    val visit_length_7s_9s_ratio = NumberUtils.formatDouble(visit_length_7s_9s / session_count, 2)
    val visit_length_10s_30s_ratio = NumberUtils.formatDouble(visit_length_10s_30s / session_count, 2)
    val visit_length_30s_60s_ratio = NumberUtils.formatDouble(visit_length_30s_60s / session_count, 2)
    val visit_length_1m_3m_ratio = NumberUtils.formatDouble(visit_length_1m_3m / session_count, 2)
    val visit_length_3m_10m_ratio = NumberUtils.formatDouble(visit_length_3m_10m / session_count, 2)
    val visit_length_10m_30m_ratio = NumberUtils.formatDouble(visit_length_10m_30m / session_count, 2)
    val visit_length_30m_ratio = NumberUtils.formatDouble(visit_length_30m / session_count, 2)
    val step_length_1_3_ratio = NumberUtils.formatDouble(step_length_1_3 / session_count, 2)
    val step_length_4_6_ratio = NumberUtils.formatDouble(step_length_4_6 / session_count, 2)
    val step_length_7_9_ratio = NumberUtils.formatDouble(step_length_7_9 / session_count, 2)
    val step_length_10_30_ratio = NumberUtils.formatDouble(step_length_10_30 / session_count, 2)
    val step_length_30_60_ratio = NumberUtils.formatDouble(step_length_30_60 / session_count, 2)
    val step_length_60_ratio = NumberUtils.formatDouble(step_length_60 / session_count, 2)
    // 将统计结果封装为Domain对象
    val sessionAggrStat = new SessionAggrStat
    sessionAggrStat.setTaskid(taskid)
    sessionAggrStat.setSession_count(session_count)
    sessionAggrStat.setVisit_length_1s_3s_ratio(visit_length_1s_3s_ratio)
    sessionAggrStat.setVisit_length_4s_6s_ratio(visit_length_4s_6s_ratio)
    sessionAggrStat.setVisit_length_7s_9s_ratio(visit_length_7s_9s_ratio)
    sessionAggrStat.setVisit_length_10s_30s_ratio(visit_length_10s_30s_ratio)
    sessionAggrStat.setVisit_length_30s_60s_ratio(visit_length_30s_60s_ratio)
    sessionAggrStat.setVisit_length_1m_3m_ratio(visit_length_1m_3m_ratio)
    sessionAggrStat.setVisit_length_3m_10m_ratio(visit_length_3m_10m_ratio)
    sessionAggrStat.setVisit_length_10m_30m_ratio(visit_length_10m_30m_ratio)
    sessionAggrStat.setVisit_length_30m_ratio(visit_length_30m_ratio)
    sessionAggrStat.setStep_length_1_3_ratio(step_length_1_3_ratio)
    sessionAggrStat.setStep_length_4_6_ratio(step_length_4_6_ratio)
    sessionAggrStat.setStep_length_7_9_ratio(step_length_7_9_ratio)
    sessionAggrStat.setStep_length_10_30_ratio(step_length_10_30_ratio)
    sessionAggrStat.setStep_length_30_60_ratio(step_length_30_60_ratio)
    sessionAggrStat.setStep_length_60_ratio(step_length_60_ratio)
    // 调用对应的DAO插入统计结果
    val sessionAggrStatDAO = DAOFactory.getSessionAggrStatDAO
    sessionAggrStatDAO.insert(sessionAggrStat)
  }

  /**
    * 生成模拟数据（只有本地模式，才会去生成模拟数据）
    *
    * @param sc
    * @param sqlContext
    */
  private def mockData(sc: SparkContext, sqlContext: SQLContext): Unit = {
    val local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL)
    if (local) MockData.mock(sc, sqlContext)
  }


}
